package cn.callmee.springboot.pulsar.starter.client.utils;

import cn.callmee.springboot.pulsar.starter.client.annotations.PulsarProducer;
import cn.callmee.springboot.pulsar.starter.client.enums.Serialization;
import cn.callmee.springboot.pulsar.starter.client.exceptions.ProducerInitException;
import cn.callmee.springboot.pulsar.starter.client.holder.Holder;
import com.google.protobuf.GeneratedMessageV3;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Schema;
import org.springframework.util.StringUtils;

import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

public class SchemaUtils {

    private SchemaUtils() {
    }

    private static <T> Schema<?> getGenericSchema(Serialization serialization, Class<T> clazz) throws RuntimeException {
        switch (serialization) {
            case JSON: {
                return Schema.JSON(clazz);
            }
            case AVRO: {
                return Schema.AVRO(clazz);
            }
            case STRING: {
                return Schema.STRING;
            }
            default: {
                throw new ProducerInitException("Unknown producer schema.");
            }
        }
    }

    private static <T extends com.google.protobuf.GeneratedMessageV3> Schema<?> getProtoSchema(Serialization serialization, Class<T> clazz) throws RuntimeException {
        if (serialization == Serialization.PROTOBUF) {
            return Schema.PROTOBUF(clazz);
        }
        throw new ProducerInitException("Unknown producer schema.");
    }

    public static Schema<?> getSchema(Serialization serialisation, Class<?> clazz) {
        if (clazz == byte[].class) {
            return Schema.BYTES;
        }

        if (isProto(serialisation)) {
            return getProtoSchema(serialisation, (Class<? extends GeneratedMessageV3>) clazz);
        }

        return getGenericSchema(serialisation, clazz);
    }

    public static boolean isProto(Serialization serialization) {
        return serialization == Serialization.PROTOBUF;
    }

    public static Class<?> getParameterType(Method method, String paramName) {
        Class<?> parameterType = getParameterType(method);
        if (StringUtils.hasText(paramName)) {
            List<Parameter> parameterList = getInvokeParameterStructure(method, paramName);
            if (parameterList.stream().anyMatch(Objects::nonNull)) {
                for (Parameter parameter : parameterList) {
                    if (null != parameter) {
                        parameterType = parameter.getType();
                        break;
                    }
                }
            }
        }
        return parameterType;
    }

    public static Class<?> getParameterType(Method method) {
        return method.getParameterTypes()[0];
    }


    private static List<Parameter> getInvokeParameterStructure(Method method, String paramName) {
        Parameter[] methodParameters = method.getParameters();
        return Arrays.stream(methodParameters).map(i -> paramName.equals(i.getName()) ? i : null).collect(Collectors.toList());
    }

    public static List<Parameter> getInvokeParameterStructure(Holder.ConsumerHolder consumerHolder) {
        Parameter[] methodParameters = consumerHolder.getHandler().getParameters();
        String paramName = consumerHolder.getAnnotation().paramName();
        if (StringUtils.hasText(paramName)) {
            return Arrays.stream(methodParameters).map(
                    i -> paramName.equals(i.getName()) || matchInvokeParameter(i) ? i : null
            ).collect(Collectors.toList());
        } else {
            return Arrays.stream(methodParameters).map(
                    i -> consumerHolder.getType().equals(i.getType()) || matchInvokeParameter(i) ? i : null
            ).collect(Collectors.toList());
        }
    }

    public static boolean matchInvokeParameter(Parameter parameter) {
        return parameter.getType().isAssignableFrom(Consumer.class)
                || parameter.getType().isAssignableFrom(MessageId.class)
                ;
    }

    // 生成生产者实际Topic列表
    public static String[] generateProducerTopicList(PulsarProducer annotationPulsarProducer) {
        List<String> topicList = Arrays.stream(annotationPulsarProducer.value()).collect(Collectors.toList());
        topicList.add(annotationPulsarProducer.topic());
        topicList.add(annotationPulsarProducer.name());
        return topicList
                .stream().filter(StringUtils::hasText)
                .collect(Collectors.toSet())
                .toArray(new String[]{});
    }
}
